package com.navinfo.platform.producer;

import com.navinfo.opentsp.platform.location.protocol.common.RealTimeDataPb;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * Created by chen on 2019/7/23.
 */
public class Producer_0F37 {
    public static String topic = "concentratedDataPB";

    public static void main(String[] args) throws InterruptedException {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "platform-010-030-050-121:9092");
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        KafkaProducer<String, byte[]> kafkaProducer = new KafkaProducer(p);

        try {
            long time = 1566061200000L;
            RealTimeDataPb.RealTimeData.Builder pb =  RealTimeDataPb.RealTimeData.newBuilder();
            pb.setTerminalId(13800138000L);
            pb.setLatitude(110000000);
            pb.setLongitude(22000000);
//            pb.setHeight(1000);
            pb.setEngineOutputTorque(100);
            pb.setSpeed(9000);
            pb.setAccelerator(5000);
            pb.setBrake(2000);
            pb.setRotation(80000);
            pb.setGear(300);
            pb.setClutchSwitch(200);
            pb.setRealTimeOilConsumption(200);
            pb.setVersion(6);
            pb.setRetransmissionPackets(100);
            pb.setFuelConsumptionRate(300);
            pb.setReceiveDate(System.currentTimeMillis()/1000);
            pb.setOriginalLat(11000);
            pb.setOriginalLng(2200);
            pb.setProtocolType(100);
            pb.setVehicleSpeed(8000);
            pb.setBarometricPressure(2000);
            pb.setFrictionTorque(2100);
            pb.setScrUpStreamNOxSensorOutput(2200);
            pb.setScrDownStreamNOxSensorOutput(2300);
            pb.setReagentAllowance(2400);
            pb.setAirInflow(100);
            pb.setScrInletTemperature(200);
            pb.setScrOutletTemperature(300);
            pb.setDpfDropoutVoltage(400);
            pb.setEngineCoolantTemperature(500);
            pb.setTankLevel(600);
            pb.setPositioningState(700);
            pb.setMileage(800);
            pb.setRValue("RRRRRRRRR");
//            pb.setSValue("SSSSSSSS");
            pb.setCommandUnit(1100);
            pb.setVin("vin123");
            pb.setTerminalVersion(1200);
            pb.setEncryption(1300);
            pb.setRLength(1400);
            pb.setSLength(1500);
            pb.setEngineBrakingState(1600);
            pb.setExhaustBrakingState(1700);
            pb.setEmptySignal(1800);
            pb.setExhaustFlow(1900);
            pb.setFanSpeed(2000);
            pb.setMechanicalBoxGear(2100);
            pb.setInstrumentSpeed(2200);
            pb.setWheelSpeed(2300);
            pb.setParkingBrakeSwitch(2400);
            pb.setCruiseControlSettingSwitch(2500);
            pb.setTargetGear(2600);
            pb.setSlope(2700);
            pb.setLoad(2800);
            pb.setFuelLevel(2900);
            pb.setWaterTemperature(3000);
            pb.setAtmosphericPressure(3100);
            pb.setIntakeAirTemperature(3200);
            pb.setAtmosphericTemperature(3300);
            pb.setExhaustGasTemperature(3400);
            pb.setIntakeManifoldBoostPressure(3500);
            pb.setRelativePressure(3600);
            pb.setEngineTorqueMode(3700);
            pb.setOilPressure(3800);
            pb.setUreaLevel(3900);
            pb.setStatusFlag(4000);
            pb.setBrakeRate(4100);
            pb.setGpsDirection(4200);
            pb.setAirConditionStatus(4300);
            pb.setTransmissionRotation(4400);
            pb.setAuxiliaryBrakeHandleStatus(1);
            pb.setUreaInjection(123000);


            for(int i=0;i<50000;i++) {
                pb.setSerialNumber(i);
                pb.setGpsTime(time+i);

                ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>(topic, pb.build().toByteArray());
                kafkaProducer.send(record);
                System.out.println("消息发送成功:" + i);
                Thread.sleep(5);
            }
        }
        catch (Exception e){
            e.printStackTrace();
        }
        finally {
            kafkaProducer.close();
        }

    }
}
